package ex55.mm.flow.impound.quote.exchange.api;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import ex55.mm.flow.impound.quote.common.kit.JsonWrapper;
import ex55.mm.flow.impound.quote.common.kit.JsonWrapperArray;
import ex55.mm.flow.impound.quote.common.stomp.StompMessage;
import ex55.mm.flow.impound.quote.common.stomp.StompMessageSerializer;
import ex55.mm.flow.impound.quote.common.stomp.TimeService;
import okhttp3.*;
import org.apache.logging.log4j.util.Strings;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class MarketDepthWSConnection extends WebSocketListener {

    private static final Logger log = LoggerFactory.getLogger(MarketDepthWSConnection.class);

    private volatile AtomicReference<ConnectionState> state = new AtomicReference<>(ConnectionState.IDLE);

    public enum ConnectionState {
        IDLE,
        DELAY_CONNECT,
        CONNECTING,
        CONNECTED,
        CLOSED_ON_ERROR
    }

    private WebSocket webSocket = null;

    private final Product product;
    private final String exchangeId;
    private volatile long lastReceivedTime = 0;
    private final int connectionId;
    private final String subscriptionUrl;

    private WebSocketWatchDog watchDog;

    private final MessageListener<MarketDataDepth> messageListener;

    private static OkHttpClient client = new OkHttpClient.Builder()
            .readTimeout(0,  TimeUnit.MILLISECONDS)
            .build();

    private static int connectionCounter = 0;

    private int delayInSecond = 0;

    public MarketDepthWSConnection(
            String exchangeId,
            Product product,
            WebSocketWatchDog dog, MessageListener<MarketDataDepth> messageListener) {
        watchDog = dog;
        this.subscriptionUrl = dog.getOptions().getWebsocketFeed();
        this.messageListener = messageListener;
        this.connectionId = MarketDepthWSConnection.connectionCounter++;
        this.product = product;
        this.exchangeId = exchangeId;
    }

    public void connect() {
        if (state.get() == ConnectionState.CONNECTED) {
            log.info("MarketDepthConnection({}.{}) Already connected", exchangeId, product.getSymbol());
            return;
        }

        if (state.get() == ConnectionState.CONNECTING) {
            log.info("MarketDepthConnection({}.{}) Connecting...", exchangeId, product.getSymbol());
            return;
        }

        this.state.set(ConnectionState.CONNECTING);

        Request request = new Request.Builder()
                .url(subscriptionUrl)
                .build();

        webSocket = client.newWebSocket(request, this);

        sendConnectMessage(webSocket);
    }

    void reConnect(int delayInSecond) {
        log.warn(
                "MarketDepthConnection({}.{}) Reconnecting after {} seconds later", exchangeId, product.getSymbol(), delayInSecond);
        if (webSocket != null) {
            closeWebSocket(webSocket);
            webSocket = null;
        }
        this.delayInSecond = delayInSecond;
        state.set(ConnectionState.DELAY_CONNECT);
    }

    private void closeWebSocket(WebSocket old) {
        if (old != null) {
            try {
                old.close(1000, "");
            } catch (Exception e) {
                log.warn("close webSocket error", e);
            }
            try {
                old.cancel();
            } catch (Exception e) {
                log.warn("cancel webSocket error", e);
            }

        }
    }

    void reConnect() {
        if (delayInSecond != 0) {
            delayInSecond--;
        } else {
            connect();
        }
    }

    public void destroy() {
        watchDog.onClosedNormally(this);
        closeWebSocket(webSocket);
        log.info("MarketDepthConnection({}.{}) destroyed", exchangeId, product.getSymbol());
    }

    @Override
    public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        log.warn("MarketDepthConnection({}.{}) is onClosing: code:{},reason:{}", exchangeId, product.getSymbol(), code, reason);
        state.set(ConnectionState.CLOSED_ON_ERROR);
        super.onClosing(webSocket, code, reason);
    }

    @Override
    public void onClosed(@NotNull WebSocket webSocket, int code, @NotNull String reason) {
        log.warn("MarketDepthConnection({}.{}) is onClosed: code:{},reason:{}", exchangeId, product.getSymbol(), code, reason);
        state.set(ConnectionState.CLOSED_ON_ERROR);
        super.onClosed(webSocket, code, reason);
    }

    @Override
    public void onFailure(@NotNull WebSocket webSocket, Throwable t, @Nullable Response response) {
        log.warn("MarketDepthConnection({}.{}) is onFailure: reason:{}", exchangeId, product.getSymbol(), t.getMessage());
        state.set(ConnectionState.CLOSED_ON_ERROR);
        super.onFailure(webSocket, t, response);
    }

    public ConnectionState getState() {
        return state.get();
    }

    long getLastReceivedTime() {
        return this.lastReceivedTime;
    }

    int getConnectionId() {
        return this.connectionId;
    }

    @Override
    public void onOpen(@NotNull WebSocket webSocket, @NotNull Response response) {
        super.onOpen(webSocket, response);
        log.info("MarketDepthConnection({}.{}) Connected to server", exchangeId, product.getSymbol());
        state.set(ConnectionState.CONNECTED);
        lastReceivedTime = TimeService.getCurrentTimeStamp();

        sendSubscribeMessage(webSocket, "/topic/orderbook/" + product.getSymbol() + "/" + exchangeId);

        watchDog.onConnectionCreated(this);
    }

    @Override public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) {

        try {
            lastReceivedTime = TimeService.getCurrentTimeStamp();

            if (text.contains("CONNECTED")) {
                return;
            }

            StompMessage message = StompMessageSerializer.deserialize(text);

            if (!Strings.isEmpty(message.getContent())) {
                JsonWrapper jsonWrapper = JsonWrapper.parseFromString(message.getContent());

                log.info("message content : {}", message.getContent());

                MarketDataDepth marketDataDepth = new MarketDataDepth();
                marketDataDepth.setProduct(product);

                JsonWrapperArray bids = jsonWrapper.getJsonArray("bids");
                if (!JsonWrapperArray.isEmpty(bids)) {
                    List<MarketDataDepth.Entry> entries = Lists.newArrayList();
                    bids.getArray().stream().limit(product.getDepth()).forEach(object -> {
                        JSONObject bid = (JSONObject) object;
                        entries.add(new MarketDataDepth.Entry().setPrice(bid.getBigDecimal("price").stripTrailingZeros()).setQuantity(bid.getBigDecimal("qty").stripTrailingZeros()));
                    });
                    marketDataDepth.setBids(entries);
                }

                JsonWrapperArray asks = jsonWrapper.getJsonArray("asks");
                if (!JsonWrapperArray.isEmpty(asks)) {
                    List<MarketDataDepth.Entry> entries = Lists.newArrayList();
                    asks.getArray().stream().limit(product.getDepth()).forEach(object -> {
                        JSONObject ask = (JSONObject) object;
                        entries.add(new MarketDataDepth.Entry().setPrice(ask.getBigDecimal("price").stripTrailingZeros()).setQuantity(ask.getBigDecimal("qty").stripTrailingZeros()));
                    });
                    marketDataDepth.setAsks(entries);
                }

                marketDataDepth.setUpdatedAt(new Date(jsonWrapper.getLong("updateAt")));

                messageListener.onMessage(marketDataDepth);
            }
        } catch (Exception ex) {
            closeWebSocket(webSocket);
        }
    }

    private void sendConnectMessage(WebSocket webSocket) {
        StompMessage message = new StompMessage("CONNECT");
        message.put("accept-version", "1.1");
        message.put("heart-beat", "10000,10000");
        webSocket.send(StompMessageSerializer.serialize(message));
    }

    private void sendSubscribeMessage(WebSocket webSocket, String topic) {
        StompMessage message = new StompMessage("SUBSCRIBE");
        message.put("id", getConnectionId() + "");
        message.put("destination", topic);
        webSocket.send(StompMessageSerializer.serialize(message));
    }

    public Product getProduct() {
        return product;
    }

    public String getExchangeId() {
        return exchangeId;
    }
}
